package util;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import util.functors.AccociBinaryOP;
import util.functors.OpScan;

public class Utilties<T> {

	
	public static <T> T  fold (final ScanDList<T> ptimes,final AccociBinaryOP<T> oplus) {
		if (ptimes == null)
			return null;
		T rst = ptimes.firstEntry().getValue();
		Entry<Long, T> entry;
		long key =ptimes.firstKey();
		while ((entry = ptimes.higherEntry(key)) != null) {
 
			key = entry.getKey();
			rst = oplus.evaluate(rst, entry.getValue());
		}
		return rst;
	}
	
	public static <T> T AccumPrevousTimes(final long chunKey,
			final ScanDList<T> ptimes, final AccociBinaryOP<T> oplus) {
		if (chunKey < 0 || ptimes == null)
			return null;
		T rst = oplus.id();

		Entry<Long, T> entry;
		long key = ptimes.firstKey();

		while (key < chunKey) {
			entry = ptimes.higherEntry(key);
			key = entry.getKey();
			rst = oplus.evaluate(rst, entry.getValue());
		}
		return rst;
	}
	
	public static class IntPairComparator implements Comparator<IntPair> {
		@Override
		public int compare(IntPair o1, IntPair o2) {
			return o1.compareTo(o2);
		}
	}

	public static class LongPairComparator implements Comparator<LongPair> {
		@Override
		public int compare(LongPair o1, LongPair o2) {
			return o1.compareTo(o2);
		}
	}

	public static Object foldl(ScanDList<?> list, AccociBinaryOP plus) {
		if (list == null || list.isEmpty())
			return null;
		long key = list.firstKey();
		Object sums = list.firstEntry().getValue();
		while (list.higherEntry(key) != null) {
			sums = plus.evaluate(sums, list.higherEntry(key).getValue());
			key = list.higherKey(key);
		}
		return sums;
	}

	public static String list2String(ArrayList<LongPair> sorted) {
		StringBuffer rst = new StringBuffer(1024);
		for (LongPair pr : sorted) {
			rst.append(pr.first + "=" + pr.second + ";");
		}
		return rst.toString();
	}

	public static void printOutput(Configuration configuration, Path file)
			throws IOException {
		FileSystem fs = FileSystem.get(configuration);
		FileStatus[] files = fs.listStatus(file);
		// byte[] buff = new byte[512];
		for (int i = 0; i < files.length; i++) {
			if (files[i].getLen() > 0) {
				FSDataInputStream in = fs.open(files[i].getPath());
				// IOUtils.readFully(in, buff , 0, 512);
				// IOUtils.copyBytes(in, System.out, 512, true);
				IOUtils.copyBytes(in, System.out, configuration, false);
				// System.out.print("/n");
				// System.out.println("...");
				in.close();
				// break;
			}

		}
		fs.delete(file, true);
	}

	// sum scan
	public static void scan(List<LongPair> list) {
		// inclusive scan
		long sum = 0;
		// int max = Integer.MAX_VALUE;
		ListIterator<LongPair> itr = list.listIterator();
		while (itr.hasNext()) {
			int idx = itr.nextIndex();
			LongPair ele = itr.next();
			list.remove(idx);
			sum = ele.second + sum;
			list.add(idx, new LongPair(ele.first, sum));
			// itr.nextIndex();
		}
	}

	// sum scan
	public static void scan(ScanList<? extends Number> list,
			OpScan<? extends Number> plus) {
		if (list.isEmpty())
			return;
		Entry<Long, ? extends Number> entry = list.firstEntry();
		Number sum = entry.getValue();
		long key = entry.getKey();
		while ((entry = list.higherEntry(key)) != null) {
			sum = plus.evaluate(sum, entry.getValue());
			key = entry.getKey();
			list.putNum(key, sum);
		}
	}

	public static void sortList(List<LongPair> list) {
		// sort
		Collections.sort(list, new LongPairComparator());
	}

	public void scan(ScanDList<T> list, AccociBinaryOP<T> plus) throws InterruptedException {
		if (list.isEmpty())
			return;
		Entry<Long, T> entry = list.firstEntry();
		T sum = entry.getValue();
		long key = entry.getKey();
		while ((entry = list.higherEntry(key)) != null) {
			sum = plus.evaluate(sum, entry.getValue());
			key = entry.getKey();
			list.put(key, sum);
		}
	}
}
